package com.mjf.day5

import java.util

import com.mjf.day3.SensorReading
import org.apache.flink.api.common.functions.{ReduceFunction, RichFlatMapFunction, RichMapFunction}
import org.apache.flink.api.common.state._
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.checkpoint.ListCheckpointed
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

/**
 * 1.什么是状态？
 *  状态是针对每个算子而言，在每个并行任务中用于计算结果的数据
 *  可以看作是一个本地变量，一般放在本地内存；flink会统一进行数据类型的管理，方便进行读写传输以及容错保证
 *
 * 2.状态分类
 *  operator state：当前任务输入的所有数据都可以访问同一份状态
 *    	List state
 *    	Union list state
 *    	Broadcast state
 *  keyed state：状态只针对当前key的数据可见，对每个key维护和管理一份状态实例（必须在keyBy之后的操作中）
 *    	ValueState[T]保存单个的值，值的类型为T
 *    	ListState[T]保存一个列表，列表里的元素的数据类型为T
 *    	MapState[K, V]保存Key-Value对
 *    	ReducingState[T]
 *    	AggregatingState[I, O]
 *
 * 3.状态后端
 *  MemoryStateBackend：内存级
 *  FsStateBackend：文件级
 *  RocksDBStateBackend：RocksDB级（数据量大时使用，支持增量）
 *
 * 4.什么样的算子可以用状态
 *  可以认为所有算子都可以有状态
 *
 * 5.什么时候是有状态的计算
 *  map/filter/flatmap本来是无状态的，但是可以通过实现RichFunction，在其中自定义状态进行操作
 *  reduce/aggregate/window本来就是有状态，是flink底层直接管理的，我们也可以实现RichFunction自定义状态
 *  Process Function是一类特殊的函数类，是.process()方法的参数，它也实现了RichFunction接口，是一个特殊的富函数
 *  DataStream/KeyedStream/ConnectedStream/WindowedStream等等都可以调用.process()方法，传入的是不同的Process Function
 */
object StateExample {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    // 配置状态后端
//    env.setStateBackend(new MemoryStateBackend()) // 内存级的状态后端,存储在TaskManager的JVM堆上
//    env.setStateBackend(new FsStateBackend("")) // 存到远程的FileSystem上，本地状态跟MemoryStateBackend一样
//    env.setStateBackend(new RocksDBStateBackend("", true))  // 将所有状态序列化后，存入本地的RocksDB中存储

    val inputStream: DataStream[SensorReading] =
      env.socketTextStream("hadoop103", 9999)
        .map {
          line =>
            val strings: Array[String] = line.split(",")
            SensorReading(strings(0), strings(1).toLong, strings(2).toDouble)
        }

    inputStream.map(new MyMap())  // operator state

    val warningStream: DataStream[(String, Double, Double)] = inputStream
      .keyBy("id")
      .flatMap(new TempChangeWarningWithFlatMap(10))  // keyed state

    warningStream.print()

    env.execute("StateExample")

  }
}


// 自定义 RichMapFunction
class TempChangeWarning(threshold: Double) extends RichMapFunction[SensorReading, (String, Double, Double)]{

  // 定义状态变量，上一次的温度值
  var lastTempState: ValueState[Double] = _

  override def open(parameters: Configuration): Unit = {
    lastTempState = getRuntimeContext.getState(new ValueStateDescriptor[Double]("last-temp", classOf[Double]))
  }

  override def map(value: SensorReading): (String, Double, Double) = {
    // 从状态中取出上次的温度值
    val lastTemp: Double = lastTempState.value()
    // 更新状态
    lastTempState.update(value.temperature)

    // 跟当前温度值计算差值，然后跟阈值比较，如果大于就报警
    val diff: Double = (value.temperature - lastTemp).abs
    if (diff > threshold) {
      (value.id, lastTemp, value.temperature)
    } else {
      null  // 会报空指针异常
    }
  }

}

// 自定义 RichFlatMapFunction,可以输出多个结果
class TempChangeWarningWithFlatMap(threshold: Double) extends RichFlatMapFunction[SensorReading, (String, Double, Double)] {

  lazy val lastTempState: ValueState[Double] = getRuntimeContext.getState(new ValueStateDescriptor[Double]("last-temp", classOf[Double]))

  override def flatMap(value: SensorReading, out: Collector[(String, Double, Double)]): Unit = {
    val lastTemp: Double = lastTempState.value()
    lastTempState.update(value.temperature)

    val diff: Double = (lastTemp - value.temperature).abs
    if(diff > threshold) {
      out.collect((value.id, lastTemp, value.temperature))
    }
  }

}


// keyed state定义示例
class MyProcessor extends KeyedProcessFunction[String, SensorReading, Int] {

  // 由于需要用到上下文环境`getRuntimeContext`,在程序编译的时候是没有上下文环境的。
  // 所以使用懒加载的方式来创建
//  lazy val myState: ValueState[Int] = getRuntimeContext.getState(new ValueStateDescriptor[Int]("my-state", classOf[Int]))

  var myState: ValueState[Int] = _
  lazy val myListState: ListState[String] = getRuntimeContext.getListState(new ListStateDescriptor[String]("my-list-state", classOf[String]))
  lazy val myMapState: MapState[String, Double] = getRuntimeContext.getMapState(new MapStateDescriptor[String, Double]("my-map-state", classOf[String], classOf[Double]))
  lazy val myReducingState: ReducingState[SensorReading] = getRuntimeContext.getReducingState(
    new ReducingStateDescriptor[SensorReading](
      "my-reducing-state",
      new ReduceFunction[SensorReading]() {
        override def reduce(value1: SensorReading, value2: SensorReading): SensorReading = {
          SensorReading(value1.id, value1.timestamp.max(value2.timestamp), value1.temperature.max(value2.temperature))
        }
      },
      classOf[SensorReading]))

  // 由于需要用到上下文环境`getRuntimeContext`,在程序编译的时候是没有上下文环境的。
  // 所以需要在初始化方法中创建（确保上下文环境存在以后）。效果同上
  override def open(parameters: Configuration): Unit = {
    myState = getRuntimeContext.getState(new ValueStateDescriptor[Int]("my-state", classOf[Int]))
  }

  override def processElement(value: SensorReading, ctx: KeyedProcessFunction[String, SensorReading, Int]#Context, out: Collector[Int]): Unit = {
    myState.value()
    myState.update(1)

    myListState.get()
    myListState.add("1")
    myListState.addAll(new util.ArrayList[String]())
    myListState.update(new util.ArrayList[String]())

    myMapState.get("sensor_1")
    myMapState.put("sensor_1", 10.1)

    myReducingState.add(value)
    myReducingState.clear()
  }

}

// operator state示例
class MyMap extends RichMapFunction[SensorReading, Long] with ListCheckpointed[java.lang.Long]{

  // 直接使用本地变量没有办法故障恢复
  var count: Long = 0L

  override def map(value: SensorReading): Long = {
    count += 1
    count
  }

  override def snapshotState(checkpointId: Long, timestamp: Long): util.List[java.lang.Long] = {
    val stateList: util.ArrayList[java.lang.Long] = new util.ArrayList[java.lang.Long]()
    stateList.add(count)
    stateList
  }

  override def restoreState(state: util.List[java.lang.Long]): Unit = {
    val iter: util.Iterator[java.lang.Long] = state.iterator()
    if (iter.hasNext) {
      count += iter.next()
    }
  }

}